package DianShang_2024.ds_01.clean

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{current_timestamp, date_format, lit, to_timestamp}

object clean02 {
  def main(args: Array[String]): Unit = {
    /*
          抽取ods库sku_info表中昨天的分区（子任务一生成的分区）数据，并结合dim_sku_info最新分区现有的数据，根据id合并数据到dwd库中dim_sku_info的
          分区表（合并是指对dwd层数据进行插入或修改，需修改的数据以id为合并字段，根据create_time排序取最新的一条），分区字段为etl_date且值与ods库的相对
          应表该值相等，并添加dwd_insert_user、dwd_insert_time、dwd_modify_user、dwd_modify_time四列,其中dwd_insert_user、dwd_modify_user
          均填写“user1”。若该条数据第一次进入数仓dwd层则dwd_insert_time、dwd_modify_time均填写当前操作时间，并进行数据类型转换。
          若该数据在进入dwd层时发生了合并修改，则dwd_insert_time时间不变，dwd_modify_time存当前操作时间，其余列存最新的值。使用hive cli查询表
          dim_sku_info的字段id、sku_desc、dwd_insert_user、dwd_modify_time、etl_date，条件为最新分区的数据，id大于等于15且小于等于20，
          并且按照id升序排序，
     */
    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .appName("数据清洗第二题")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd")

    //sku_info_temp:首先对ods.sku_info2表进行添加对应的列名操作
    spark.sql("select *  from ods.sku_info2 where etl_date='20231017'")
      .drop("etl_date")
      .withColumn("dwd_insert_user",lit("user1"))
      .withColumn(
        "dwd_insert_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )
      .withColumn("dwd_modify_user",lit("user1"))
      .withColumn(
        "dwd_modify_time",
        lit(to_timestamp(date_format(current_timestamp(),"yyyy-MM-dd HH:mm:ss"),"yyyy-MM-dd HH:mm:ss"))
      )
      .withColumn("etl_date",lit("20231017"))
      .createOrReplaceTempView("sku_info_temp")



    //  t1:然后对dwd.dim_sku_info和ods.sku_info2进行合并操作
    spark.sql(
      """
        |select * from dwd.dim_sku_info
        |union all
        |select * from sku_info_temp
        |""".stripMargin).createOrReplaceTempView("t1")

    //  t2:对合并的数据进行窗口内排序(根据id分区,根据create_time排序)
    spark.sql(
      """
        |select
        |id,
        |spu_id,
        |price,
        |sku_name,
        |sku_desc,
        |weight,
        |tm_id,
        |category3_id,
        |sku_default_img,
        |create_time,
        |row_number() over(partition by id order by create_time desc) as row,
        |dwd_insert_user,
        |min(dwd_insert_time) over(partition by id) as dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from
        |t1
        |""".stripMargin).createOrReplaceTempView("t2")

    spark.sql("select * from t2 where row=1 ").show


    //  插入每个窗口排在第一行的数据
    spark.sql(
      """
        |insert into table dwd.dim_sku_info
        |partition(etl_date='20231017')
        |select
        |id,
        |spu_id,
        |price,
        |sku_name,
        |sku_desc,
        |weight,
        |tm_id,
        |category3_id,
        |sku_default_img,
        |create_time,
        |dwd_insert_user,
        |dwd_insert_time,
        |dwd_modify_user,
        |dwd_modify_time
        |from t2
        |where row=1
        |""".stripMargin)


    //  关闭sparksql的环境
    spark.close()

  }

}
